package org.breathe.aof.loader;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.breathe.command.Command;
import org.breathe.command.CommandType;
import org.breathe.protocal.Resp;
import org.breathe.protocal.decode.BulkString;
import org.breathe.protocal.decode.RespArray;
import org.breathe.server.core.RedisCore;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * @author: breathe
 * @createTime: 2025-06-08
 */
@Slf4j
public class AofLoader {
    private static final byte[] REDIS_PREFIX = new byte[]{'*', '$', '+', '-', ':'};
    private final String fileName;
    private final RedisCore redisCore;
    private FileChannel fileChannel;
    private RandomAccessFile randomAccessFile;
    private static final int BUFFER_SIZE = 4096;

    public AofLoader(String fileName, RedisCore redisCore) throws Exception {
        this.fileName = fileName;
        this.redisCore = redisCore;
        openFile();
    }
    public void openFile() throws IOException {
        File file = new File(fileName);
        if (!file.exists() || file.length() == 0) {
            log.info("AOF file not exists");
            return ;
        }
        this.randomAccessFile = new RandomAccessFile(file, "r");
        this.fileChannel = randomAccessFile.getChannel();
    }

    public void load() {
        if (fileChannel == null) {
            log.info("fileChannel is null");
            return ;
        }
        try {
            log.info("start load aof file");
            ByteBuf commands = readFileContent();
            int successCount = processCommands(commands);
            log.info("load aof file successfully, have {} commands", successCount);
        } catch (Exception ex) {
            log.error("load aof file fail", ex);
            throw new RuntimeException(ex);
        } finally {
            closeFile();
        }

    }

    private int processCommands(ByteBuf commands) {
        int successCount = 0;
        while (commands.isReadable()) {
            int position = commands.readerIndex();
            commands.markReaderIndex();
            try {
                Resp command = Resp.decode(commands);
                if (executeCommand(command, position)) {
                    successCount ++;
                }
            } catch (Exception e) {
                handleCommandError(commands, position, e);
            }
        }
        return successCount;
    }

    private void handleCommandError(ByteBuf commands, int position, Exception e) {
        log.warn("command run fail, at {}", position, e.getMessage());

        commands.resetReaderIndex();
        while (commands.isReadable()) {
            byte b = commands.readByte();
            if (isRespPrefix(b)) {
                commands.readerIndex(commands.readerIndex() - 1);
                break;
            }
        }
    }

    private boolean isRespPrefix(byte b) {
        for (byte prefix : REDIS_PREFIX) {
            if (prefix == b) {
                return true;
            }
        }
        return false;
    }

    private boolean executeCommand(Resp respArray, int position) {
        if (!(respArray instanceof RespArray)) {
            log.warn("command's type is not same to respArray, at {}", position);
            return false;
        }
        RespArray command = (RespArray) respArray;
        if (!isValidCommand(command)) {
            log.warn("command run fail");
            return false;
        }
        return executeRedisCommand(command, position);
    }

    private boolean executeRedisCommand(RespArray command, int position) {
        try {
            String commandName = ((BulkString)command.getContent()[0]).getContent().getString().toUpperCase();
            CommandType commandType = null;
            try {
                commandType = CommandType.valueOf(commandName);
            } catch (IllegalArgumentException e) {
                return false;
            }
            Command cmd = commandType.getSupplier().apply(redisCore);
            cmd.setContext(command.getContent());
            Resp result = cmd.handle();
            return true;
        } catch (Exception ex) {
            log.error("command run fail, at {}", position, ex);
        }
        return false;
    }

    private boolean isValidCommand(RespArray command) {
        Resp[] content = command.getContent();
        return content.length > 0 && content[0] instanceof BulkString;
    }

    private ByteBuf readFileContent() throws IOException{
        long fileSize = fileChannel.size();
        if (fileSize == 0) {
            return Unpooled.EMPTY_BUFFER;
        }
        if (fileSize > Integer.MAX_VALUE) {
            return readLargeFile();
        }
        int size = (int) fileSize;
        ByteBuf buffer = PooledByteBufAllocator.DEFAULT.directBuffer(size);
        try {
            ByteBuffer byteBuffer = buffer.nioBuffer(0, size);
            int totalRead = 0;
            while (totalRead < size) {
                int read = fileChannel.read(byteBuffer);
                if (read == -1) {
                    break;
                }
                totalRead += read;
            }
            buffer.writerIndex(totalRead);
            return buffer;
        } catch (IOException ex) {
            buffer.release();
            throw ex;
        }
    }

    private ByteBuf readLargeFile() {
        CompositeByteBuf compositeByteBuf = PooledByteBufAllocator.DEFAULT.compositeBuffer();
        ByteBuf currentBuf = null;
        try {
            long remaining = fileChannel.size();
            while (remaining > 0) {
                int chunkSize = Math.min(BUFFER_SIZE, (int) remaining);
                currentBuf = PooledByteBufAllocator.DEFAULT.directBuffer(chunkSize);
                ByteBuffer byteBuffer = currentBuf.nioBuffer(0, chunkSize);
                int read = fileChannel.read(byteBuffer);
                if (read == -1) {
                    currentBuf.release();
                    break;
                }
                currentBuf.writerIndex(read);
                compositeByteBuf.addComponent(currentBuf);
                remaining -= read;
                currentBuf = null;
            }
            return compositeByteBuf;
        } catch (IOException ex) {
            if (currentBuf != null) {
                currentBuf.release();
            }
            compositeByteBuf.release();
            throw new RuntimeException(ex);
        }
    }
    public void closeFile(){
        try{
            if(fileChannel != null) {
                fileChannel.close();
            }
            if(randomAccessFile != null) {
                randomAccessFile.close();
            }
        }catch(IOException e){
            log.error("error when close aofLoader",e);
        }
    }


    public void close(){
        closeFile();
    }
}
